package cn.ywang.current.five;

import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 阻塞队列
 *
 * @author BeatificWang
 * @date 2019/7/28 16:54
 */
public class BlockingQueue<E> {

	/**
	 * 节点元素
	 */
	static class Node<E> {
		E item;

		Node<E> next;

		Node(E item) {
			this.item = item;
		}
	}


	public BlockingQueue(int capacity) {
		if (capacity <= 0) {
			throw new IllegalArgumentException();
		}
		this.capacity = capacity;
		this.tail = this.head = new Node<>(null);
	}


	/**
	 * 取元素
	 *
	 * @return
	 * @throws InterruptedException
	 */
	public E take() throws InterruptedException {
		E x;
		int c = -1;
		final AtomicInteger count = this.count;
		final ReentrantLock takeLock = this.takeLock;
		takeLock.lockInterruptibly();
		try {
			while (count.get() == 0) {
				this.notEmpty.await();
			}
			x = this.dequeue();
			c = count.getAndDecrement();
			if (c > 1) {
				this.notEmpty.signal();
			}
		} finally {
			takeLock.unlock();
		}
		if (c == this.capacity) {
			this.signalNotFull();
		}
		return x;
	}

	/**
	 * 入队
	 *
	 * @param e
	 */
	public void put(E e) throws InterruptedException {
		if (e == null) {
			throw new NullPointerException();
		}
		// Note: convention in all put/take/etc is to preset local var
		// holding count negative to indicate failure unless set.
		int c = -1;
		Node<E> node = new Node<E>(e);
		final ReentrantLock putLock = this.putLock;
		final AtomicInteger count = this.count;
		putLock.lockInterruptibly();
		try {
			while (count.get() == this.capacity) {
				this.notFull.await();
			}
			this.enqueue(node);
			c = count.getAndIncrement();
			if (c + 1 < this.capacity) {
				this.notFull.signal();
			}
		} finally {
			putLock.unlock();
		}
		if (c == 0) {
			this.signalNotEmpty();
		}

	}


	/**
	 * 容量
	 */
	private final int capacity;

	/**
	 * 元素个数
	 */
	private final AtomicInteger count = new AtomicInteger();

	/**
	 * 队头
	 */
	transient Node<E> head;

	/**
	 * 队尾
	 */
	private transient Node<E> tail;

	/**
	 * Lock held by take, poll, etc
	 */
	private final ReentrantLock takeLock = new ReentrantLock();

	/**
	 * Wait queue for waiting takes
	 */
	private final Condition notEmpty = this.takeLock.newCondition();

	/**
	 * Lock held by put, offer, etc
	 */
	private final ReentrantLock putLock = new ReentrantLock();

	/**
	 * Wait queue for waiting puts
	 */
	private final Condition notFull = this.putLock.newCondition();

	private E dequeue() {
		Node<E> h = this.head;
		Node<E> first = h.next;
		// help GC
		h.next = h;
		this.head = first;
		E x = first.item;
		first.item = null;
		return x;
	}

	/**
	 * 入队
	 *
	 * @param node 节点值
	 */
	private void enqueue(Node node) {
		this.tail = this.tail.next = node;
	}

	/**
	 * 非空时唤醒线程
	 */
	private void signalNotEmpty() {
		final ReentrantLock takeLock = this.takeLock;
		takeLock.lock();
		try {
			this.notEmpty.signal();
		} finally {
			takeLock.unlock();
		}
	}

	private void signalNotFull() {
		final ReentrantLock putLock = this.putLock;
		putLock.lock();
		try {
			this.notFull.signal();
		} finally {
			putLock.unlock();
		}
	}
}
